package com.xzy.rabbit.demo.delayqueue.c;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 延迟队列——消息TTL
 *
 * @author xzy
 * @date 2021/9/2914:59
 */
@Slf4j
@AllArgsConstructor
class Producer implements Runnable {
    private final Connection connection;
    private final String exchangeName;
    private final String routingKey;
    private final Integer messageCount;
    private final long workTime;
    private final long messageDelayTime;

    public void produce() throws IOException, TimeoutException, InterruptedException {
        // 1.获取信道
        try (Channel channel = connection.createChannel()) {

            // 2.发送消息

            // 延迟时间
            Map<String, Object> headers = new HashMap<>();
            headers.put("x-delay", messageDelayTime);

            AMQP.BasicProperties properties = new AMQP.BasicProperties()
                    .builder()
                    .headers(headers)
                    .build();

            String currentThread = Thread.currentThread().getName();
            for (int i = 0; i < messageCount; i++) {
                Thread.sleep(workTime);
                String message = messageDelayTime + "ms-msg" + i;
                channel.basicPublish(exchangeName, routingKey, properties, message.getBytes());
                log.info(currentThread + " 成功发送：`{}`----[{}][{}ms]---->\n", message, routingKey, messageDelayTime);
            }
        }
    }

    @Override
    public void run() {
        try {
            produce();
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        }
    }
}
